import java.io.IOException;
import java.util.Set;
import java.util.concurrent.TimeoutException;

import com.google.gson.Gson;
import com.google.gson.JsonSyntaxException;
import com.jiuan.transfer.mqtt.MqttUtil;
import com.jiuan.transfer.mqtt.domain.MsgBean;
import com.jiuan.transfer.redis.SingleRedisUtil;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
/**
 * Created by Administrator on 2017/1/4.
 */
public class WorkThread extends Thread{
    private String queue;
    private Channel channel;
    private Gson gson;
    public WorkThread(String queue, Channel channel) {
        this.queue = queue;
        this.channel = channel;
        gson=new Gson();
    }



    // 发布
    private static void publish(String topic, Set<String> queryIps,
                                byte[] message) {
        if(null==message || message.length<=0)return;
        if(null==queryIps || queryIps.size()<=0)return;
        if(null==topic || "".equals(topic))return;

        for (String ip : queryIps) {
            try {
                MqttUtil.publish(ip, topic, message);
            } catch (Exception e) {

                e.printStackTrace();
            }
        }
    }




    public void run() {
        try {
            System.out.println(Thread.currentThread().getName()+"----------"+"WorkThread");
            channel.queueDeclare("all.mqtt_publish", true, false, false, null);
            boolean autoAck = false;
            channel.basicConsume(this.queue, autoAck, new DefaultConsumer(
                    channel) {
                @Override
                public void handleDelivery(String consumerTag,
                                           Envelope envelope, AMQP.BasicProperties properties,
                                           byte[] body) throws IOException {

                    System.out.println(Thread.currentThread().getName()+"----------"+"handleDelivery");

                    long deliveryTag = envelope.getDeliveryTag();

                    // 收到message
                    String message = new String(body, "UTF-8");
                    MsgBean msgBean = null;

                    try {
                        msgBean = gson.fromJson(message, MsgBean.class);

                    } catch (JsonSyntaxException e) {
                        e.printStackTrace();
                        System.out.println("消息格式錯誤");
                        return;
                    }
                    System.out.println("RabbitMQ   [x] Received:     " + msgBean);

                    // 獲取主題和消息内容
                    String topic = msgBean.getTopic();
                    byte[] publishmessage = (gson.toJson(msgBean)).getBytes();


                    System.out.println(gson.toJson(msgBean));

                    // 查询推送ip
                    Set<String> ips = SingleRedisUtil.queryIps("TOPIC_" + topic);


                    // 发布主题MQQT
                    publish(topic, ips, publishmessage);

                    channel.basicAck(deliveryTag, false);

                }

            });
            System.out.println(Thread.currentThread().getName()
                    + ">>   RabbitMq 订阅OK!");
        } catch (IOException e) {
            System.out.println("RabbitMq 订阅失败");
            e.printStackTrace();
            try {
                channel.close();
            } catch (IOException | TimeoutException e1) {

            }
        }
    }



}
